package com.charles.bigdata.mapreduce.joinuser;

import java.io.IOException;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.jobcontrol.ControlledJob;
import org.apache.hadoop.mapreduce.lib.jobcontrol.JobControl;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import com.charles.bigdata.mapreduce.joinuser.StepFirst.FirstMapper;
import com.charles.bigdata.mapreduce.joinuser.StepFirst.FirstReducer;
import com.charles.bigdata.mapreduce.joinuser.StepSecond.SecondMapper;
import com.charles.bigdata.mapreduce.joinuser.StepSecond.SecondReducer;

public class JobControlMain {
	// 启动函数
	public static void main(String[] args) throws IOException {

		JobConf conf = new JobConf(JobControlMain.class);

		// 第一个job的配置
		Job job1 = Job.getInstance(conf);
		job1.setJarByClass(StepFirst.class);

		job1.setMapperClass(FirstMapper.class);
		job1.setReducerClass(FirstReducer.class);

		job1.setOutputKeyClass(Text.class);
		job1.setOutputValueClass(Text.class);

		// 加入控制容器
		ControlledJob ctrljob1 = new ControlledJob(conf);
		ctrljob1.setJob(job1);

		// job1的输入输出文件路径
		FileInputFormat.setInputPaths(job1, new Path(args[0]));
		FileOutputFormat.setOutputPath(job1, new Path(args[1]));

		// 第二个作业的配置
		Job job2 = Job.getInstance(conf);
		job2.setJarByClass(StepSecond.class);

		job2.setMapperClass(SecondMapper.class);
		job2.setReducerClass(SecondReducer.class);

		job2.setOutputKeyClass(Text.class);
		job2.setOutputValueClass(Text.class);

		// 作业2加入控制容器
		ControlledJob ctrljob2 = new ControlledJob(conf);
		ctrljob2.setJob(job2);

		// 设置多个作业直接的依赖关系
		// 如下所写：
		// 意思为job2的启动，依赖于job1作业的完成
		ctrljob2.addDependingJob(ctrljob1);

		// 输入路径是上一个作业的输出路径，因此这里填args[1],要和上面对应好
		FileInputFormat.addInputPath(job2, new Path(args[1]));

		// 输出路径从新传入一个参数，这里需要注意，因为我们最后的输出文件一定要是没有出现过得
		// 因此我们在这里new Path(args[2])因为args[2]在上面没有用过，只要和上面不同就可以了
		FileOutputFormat.setOutputPath(job2, new Path(args[2]));

		// 主的控制容器，控制上面的总的两个子作业
		JobControl jobCtrl = new JobControl("myctrl");

		// 添加到总的JobControl里，进行控制
		jobCtrl.addJob(ctrljob1);
		jobCtrl.addJob(ctrljob2);

		// 在线程启动，记住一定要有这个
		Thread t = new Thread(jobCtrl);
		t.start();

		while (true) {

			if (jobCtrl.allFinished()) {// 如果作业成功完成，就打印成功作业的信息
				System.out.println(jobCtrl.getSuccessfulJobList());
				jobCtrl.stop();
				break;
			}
		}
	}
}
